package com.superiron.scaffold.core.service.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author ：yzg
 * @date ：Created in 2022/8/29 14:04
 * @description：
 * @modified By：
 * @version: $
 */
@Component
@Order(value = 9)
@Slf4j
public class InitMqProducer implements ApplicationListener<ApplicationReadyEvent> {
	
	
	@Autowired
	private MyRocketMqProperties myRocketMqProperties;
	
	/**
	 * 存放所有生产者
	 */
	public static Map<String, DefaultMQProducer> producerMap = new HashMap<>();
	
	@Override
	public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
		
		if (myRocketMqProperties.isEnable()) {
			
			List<MqProducerVo> producerlist = myRocketMqProperties.getProducerlist();
			if (CollectionUtils.isEmpty(producerlist)) {
				log.info("无MQ生产者---------------------------------------");
			}
			
			for (MqProducerVo vo : producerlist) {
				try {
					DefaultMQProducer defaultMQProducer = new DefaultMQProducer(vo.getProducerId());
					defaultMQProducer.setNamesrvAddr(vo.getNamesrvAddr());
					defaultMQProducer.setVipChannelEnabled(false);
					defaultMQProducer.setInstanceName(vo.getProducerId());
					//客户端回调线程数，默认：Runtime.getRuntime.availableProcessors，当前cpu核心数
					defaultMQProducer.setClientCallbackExecutorThreads(4);
					//持久化消费者时间间隔
					defaultMQProducer.setPersistConsumerOffsetInterval(5000);
					defaultMQProducer.setSendMsgTimeout(10000);//生产发送消息超时时间
					defaultMQProducer.setCompressMsgBodyOverHowmuch(4096); //消费体容量上限，默认是4m
					defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(false); //是否在内部发送失败时重试另一个broker
					defaultMQProducer.setRetryTimesWhenSendFailed(2);//同步模式下重试次数限制
					defaultMQProducer.setRetryTimesWhenSendAsyncFailed(2); //异步模式下重试次数限制
					defaultMQProducer.start();
					
					producerMap.put(vo.getProducerId(), defaultMQProducer);
					log.info("mq生产者{},{}启动成功", vo.getProducerId(), vo.getNamesrvAddr());
				} catch (MQClientException e) {
					log.error("mq生产者{},{}启动失败", vo.getProducerId(), vo.getNamesrvAddr(), e);
				}
			}
		}
		
	}
	
}

